package bjut.core.serializal;

import bjut.core.Message.RpcRequest;
import bjut.core.Message.RpcResponse;
import bjut.core.interfaces.Serializer;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import org.yaml.snakeyaml.serializer.SerializerException;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;

/**
 * @Author: zhuzhiming
 * @Date: 2022/4/12 0:35
 */
public class KryoSerializer implements Serializer {

    /**
     * Because Kryo is not thread safe. So, use ThreadLocal to store Kryo objects
     */
    private final ThreadLocal<Kryo> kryoThreadLocal = ThreadLocal.withInitial(() -> {
        Kryo kryo = new Kryo();
        //kryo在序列化对象时，首先会序列化其类的全限定名，由于我们通常序列化的对象都是有限范围内的类的实例，这样重复序列化同样的类的全限定名是低效的。
        // 通过注册kryo可以将类的全限定名抽象为一个数字，即用一个数字代表全限定名，这样就要高效一些。
        kryo.register(RpcResponse.class);
        kryo.register(RpcRequest.class);
        return kryo;
    });

    @Override
    public byte[] serialize(Object obj) {

        try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream()) {
            Output output = new Output(byteArrayOutputStream);
            Kryo kryo = kryoThreadLocal.get();
            kryo.writeObject(output, obj);
            kryoThreadLocal.remove();
            return output.toBytes();
        } catch (IOException e) {
            throw new SerializerException("serializer exception");
        }
    }

    @Override
    public <T> T deserialize(byte[] bytes, Class<T> cls) {

        try (ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes)) {
            Input input = new Input(byteArrayInputStream);
            Kryo kryo = kryoThreadLocal.get();
            T t = kryo.readObject(input, cls);
            kryoThreadLocal.remove();
            return cls.cast(t);
        } catch (IOException e) {
            throw new SerializerException("deserializer  exception");
        }

    }
}
